package com.bjbn.service;

import com.bjbn.bean.entity.FltInfoDynamic;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class FlightDataProcessor {

    @Autowired
    private FlightAirlineService flightAirlineService;

    @Autowired
    private FltInfoDynamicService fltInfoDynamicService;
    
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Value("${flight.data.kafka.target-topic}")
    private String targetTopic;

    // 延迟队列，用于处理需要重试的消息
    private final DelayQueue<DelayedMessage> delayQueue = new DelayQueue<>();

    // 从配置文件读取重试配置
    @Value("${flight.data.retry.max-count:10}")
    private int maxRetryCount;

    @Value("${flight.data.retry.interval-ms:500}")
    private long retryIntervalMs;

    public FlightDataProcessor() {
        // 启动延迟队列处理线程
        startDelayQueueProcessor();
    }

    /**
     * 处理航班航站信息
     */
    @Async("flightDataTaskExecutor")
    public CompletableFuture<Void> processAirlineInfo(String message) {
        return CompletableFuture.runAsync(() -> {
            try {
                processAirlineInfoWithRetry(message, 0);
            } catch (Exception e) {
                log.error("处理航班航站信息失败: " + message + ", " + e.getMessage(), e);
            }
        });
    }

    /**
     * 带重试的航班航站信息处理
     */
    private void processAirlineInfoWithRetry(String message, int retryCount) {
        try {
            FltInfoDynamic flightInfo = FlightHandle.handleFltAirline(message, flightAirlineService);
            
            if (flightInfo != null) {
                log.debug("FlightInfo: " + flightInfo);
                fltInfoDynamicService.saveOrUpdateFltInfoDynamic(flightInfo);
                // 将航班动态数据推送到Kafka
                sendToKafka(flightInfo);
                log.debug("航班航站信息处理成功，重试次数: " + retryCount);
            } else {
                if (retryCount < maxRetryCount) {
                    log.debug("航班航站数据未就绪，第" + retryCount + "次重试，等待" + retryIntervalMs + "ms");
                    // 将消息放入延迟队列，稍后重试
                    delayQueue.offer(new DelayedMessage(message, retryCount + 1, retryIntervalMs));
                } else {
                    log.warn("航班航站数据处理失败，已达到最大重试次数: " + maxRetryCount);
                }
            }
        } catch (Exception e) {
            log.error("处理航班航站信息异常: " + message + ", " + e.getMessage(), e);
        }
    }

    /**
     * 启动延迟队列处理线程
     */
    private void startDelayQueueProcessor() {
        Thread processorThread = new Thread(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    DelayedMessage delayedMessage = delayQueue.take();
                    processAirlineInfoWithRetry(delayedMessage.getMessage(), delayedMessage.getRetryCount());
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.info("延迟队列处理线程被中断");
                    break;
                } catch (Exception e) {
                    log.error("延迟队列处理异常: " + e.getMessage(), e);
                }
            }
        });
        processorThread.setName("FlightData-DelayProcessor");
        processorThread.setDaemon(true);
        processorThread.start();
        log.info("延迟队列处理线程已启动");
    }

    /**
     * 获取延迟队列大小
     */
    public int getDelayQueueSize() {
        return delayQueue.size();
    }

    /**
     * 延迟消息内部类
     */
    private static class DelayedMessage implements Delayed {
        private final String message;
        private final int retryCount;
        private final long executeTime;

        public DelayedMessage(String message, int retryCount, long delayMs) {
            this.message = message;
            this.retryCount = retryCount;
            this.executeTime = System.currentTimeMillis() + delayMs;
        }

        public String getMessage() {
            return message;
        }

        public int getRetryCount() {
            return retryCount;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(executeTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }

        @Override
        public int compareTo(Delayed other) {
            return Long.compare(executeTime, ((DelayedMessage) other).executeTime);
        }
    }
    
    /**
     * 将航班动态数据发送到Kafka
     * @param flightInfo 航班动态信息
     */
    private void sendToKafka(FltInfoDynamic flightInfo) {
        try {
            // 使用航班ID作为消息键，确保相同航班的数据分区一致
            String key = flightInfo.getFLID().toString();
            kafkaTemplate.send(targetTopic, key, flightInfo);
            log.debug("Successfully sent flight data to Kafka topic {}. Flight ID: {}", targetTopic, key);
        } catch (Exception e) {
            log.error("Failed to send flight data to Kafka. Error: {}", e.getMessage(), e);
        }
    }
}